<?php

namespace App\Http\Common\amqp;

use App\Http\Common\amqp\mode\DirectMode;
use common\amqp\exceptions\AmqpException;
use Illuminate\Support\Facades\Log;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exchange\AMQPExchangeType;

abstract class RabbitMQ
{
    /**
     * @var AMQPStreamConnection
     */
    protected $connection;

    /**
     * @var AMQPChannel
     */
    protected $channel;

    /**
     * @var string
     */
    protected $queueName = '';

    /**
     * @var string
     */
    protected $exchangeName = '';

    /**
     * @var string
     */
    protected $routeKey = '';

    /**
     * @var string
     */
    protected $exchangeType = AMQPExchangeType::DIRECT;

    /**
     * @var string 链接key
     */
    protected $connectKey;

    /**
     * @param array $msgBody
     * @return bool
     */
    public abstract function sendMsg(array $msgBody): bool;

    /**
     * @param callable $callback
     * @return void
     */
    public abstract function receiveMsg(callable $callback): void;

    public function connection()
    {

    }

    public function getClient()
    {
        if (empty($this->connection)) {
            $configs = config('amqp')[$this->connectKey];
            if (empty($configs)) {
                throw new \Exception('配置不能为空');
            }
            $this->connection = new AMQPStreamConnection(
                $configs['host'],
                $configs['port'],
                $configs['user'],
                $configs['passwd'],
                $configs['vhost']
            );
        }
        return $this->connection;
    }

    /**
     * @return \PhpAmqpLib\Channel\AbstractChannel|AMQPChannel
     * @throws \Exception
     */
    public function getChannel()
    {
        if (empty($this->channel)) {
            $this->channel = $this->getClient()->channel();
        }
        return $this->channel;
    }

    /**
     * @param string $connectKey
     * @return $this
     */
    public function setConnectKey(string $connectKey): self
    {
        $this->connectKey = $connectKey;
        return $this;
    }

    /**
     * @param string $exchangeName
     * @return $this
     */
    public function setExchangeName(string $exchangeName): self
    {
        $this->exchangeName = $exchangeName;
        return $this;
    }

    /**
     * @param string $queueName
     * @return $this
     */
    public function setQueueName(string $queueName): self
    {
        $this->queueName = $queueName;
        return $this;
    }

    /**
     * @param string $routeKey
     * @return $this
     */
    public function setRouteKey(string $routeKey): self
    {
        $this->routeKey = $routeKey;
        return $this;
    }

    /**
     * 删除消息 （手动应答）
     * @param $msg
     * @return void
     */
    public function deleteMessage($msg): void
    {
        /** @var AMQPChannel $channel */
        $channel = $msg->delivery_info['channel'];
        $channel->basic_ack($msg->delivery_info['delivery_tag']);
    }

    public function sendCheck()
    {
        if (empty($this->exchangeName)) {
            throw new \Exception('请指定交换机');
        }
    }

    /**
     * @throws \Exception
     */
    protected function receiveCheck()
    {
        if (empty($this->exchangeName)) {
            throw new \Exception('请指定交换机');
        }
        if (empty($this->queueName)) {
            throw new \Exception('请指定队列');
        }
        if (empty($this->routeKey)) {
            throw new \Exception('请指定路由');
        }
    }

    /**
     * @return void
     */
    protected function wait()
    {
        while (count($this->channel->callbacks)) {
            try {
                $this->channel->wait();
            } catch (\Exception $exception) {
                Log::info('AmqpWaitError：' . $exception->getMessage(), [
                    'exception' => $exception
                ]);
                break;
            }
        }
    }

    protected function closeChannel()
    {
        try {
            // 关闭通道
            !empty($this->channel) && $this->channel->close();
        } catch (\Exception $exception) {
            Log::info('channel close error: ' . $exception->getMessage(), [
                'exception' => $exception
            ]);
        }
    }


}
